๐ rns.kin.earth git ๐
Commit b9278ce352332a16508b2927e404e4f5dda806e8
Parents : 0e070aa
Author : Mark Qvist <bc7291552be7a58f361522990465165c>
Signature : T66BB85Valid, signed by author
Date : 2026-08-24T20:42:57+02:00
Added throughput benchmarker
Changes
Diff
diff --git a/tests/throughput.py b/tests/throughput.py
new file mode 100644
index 00000000..7112eaf0
--- /dev/null
+++ b/tests/throughput.py
@@ -0,0 +1,883 @@
+#!/usr/bin/env python3
+#
+# Usage examples:
+#
+# python3 tests/transport_throughput.py
+# python3 tests/transport_throughput.py --scenario transit_single_135
+# python3 tests/transport_throughput.py --mode inline --runs 5
+# python3 tests/transport_throughput.py --list-scenarios
+#
+
+import unittest
+
+import os
+import sys
+import time
+import gc
+import struct
+import platform
+import tempfile
+import statistics
+
+# Ensure that the Reticulum tree this suite lives in is the one being
+# benchmarked, even if a different version of Reticulum is installed in
+# site-packages.
+_SUITE_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+if os.path.isdir(os.path.join(_SUITE_ROOT, "RNS")):
+ sys.path.insert(0, _SUITE_ROOT)
+
+import RNS
+
+# ---------------------------------------------------------------------------
+# Configuration
+# ---------------------------------------------------------------------------
+
+BENCHMARK_CONFIG = {
+ "scenarios": None, # None = all, or a list of scenario names
+ "mode": "both", # "inline", "drainer" or "both"
+ "runs": 3, # measurement runs per scenario/mode, median is reported
+}
+
+SCENARIO_DESCRIPTIONS = {
+ "transit_single_135": "SINGLE transit relay, 135 B payload, 3-hop path",
+ "transit_single_475": "SINGLE transit relay, 475 B payload, 3-hop path",
+ "transit_single_1024": "SINGLE transit relay, 1 KiB payload, 3-hop path",
+ "transit_single_16384": "SINGLE transit relay, 16 KiB payload, 3-hop path",
+ "transit_single_final": "SINGLE transit relay, final hop, header strip, 135 B",
+ "transit_link_135": "LINK transit relay via link table, cross-interface, 135 B",
+ "transit_link_475": "LINK transit relay via link table, cross-interface, 475 B",
+ "transit_link_1024": "LINK transit relay via link table, cross-interface, 1024 B",
+ "transit_link_16384": "LINK transit relay via link table, cross-interface, 16384 B",
+ "terminus_link": "LINK terminus delivery, token decrypt, 135 B",
+ "terminus_single": "SINGLE local delivery, ephemeral-key decrypt, 135 B",
+ "announce_ingress": "Announce ingress, fresh destinations, validation + path insert",
+ "outbound_path": "Outbound insertion into transport, known 3-hop path, 135 B",
+}
+
+# Default packet counts per scenario and mode.
+DEFAULT_PACKETS = {
+ "transit_single_135": {"inline": 20000, "drainer": 20000},
+ "transit_single_475": {"inline": 20000, "drainer": 20000},
+ "transit_single_1024": {"inline": 20000, "drainer": 20000},
+ "transit_single_16384": {"inline": 10000, "drainer": 10000},
+ "transit_single_final": {"inline": 20000, "drainer": 20000},
+ "transit_link_135": {"inline": 20000, "drainer": 20000},
+ "transit_link_475": {"inline": 20000, "drainer": 20000},
+ "transit_link_1024": {"inline": 20000, "drainer": 20000},
+ "transit_link_16384": {"inline": 10000, "drainer": 10000},
+ "terminus_link": {"inline": 6000, "drainer": 6000},
+ "terminus_single": {"inline": 3000, "drainer": 3000},
+ "announce_ingress": {"inline": 1000, "drainer": 1000},
+ "outbound_path": {"inline": 8000, "drainer": 0},
+}
+
+INLINE_SAMPLE_STEP = 64 # sample a timing point every N packets for p50/p95
+DRAINER_CHUNK = 256 # packets fed per drainer backpressure step
+
+
+class BenchmarkInterface(RNS.Interfaces.Interface.Interface):
+ HW_MTU = 1048576
+ BITRATE = 1_000_000_000
+
+ def __init__(self, name="bench"):
+ super().__init__()
+ self.name = name
+ self.IN = True
+ self.OUT = True
+ self.online = True
+ self.bitrate = self.BITRATE
+ self.HW_MTU = BenchmarkInterface.HW_MTU
+ self.mode = RNS.Interfaces.Interface.Interface.MODE_FULL
+ self.gravity = 0
+ self.ifac_size = 0
+ self.ifac_key = None
+ self.ifac_identity = None
+ self.parent_interface = None
+ self.ingress_control = False
+ self.announce_rate_target = RNS.Interfaces.Interface.Interface.DEFAULT_AR_TARGET
+ self.announce_rate_grace = RNS.Interfaces.Interface.Interface.DEFAULT_AR_GRACE
+ self.announce_rate_penalty = RNS.Interfaces.Interface.Interface.DEFAULT_AR_PENALTY
+
+ def process_outgoing(self, data):
+ self.txb += len(data)
+
+ def __str__(self):
+ return f"BenchmarkInterface[{self.name}]"
+
+
+class Scenario:
+ kind = "ingress" # "ingress" or "outbound"
+
+ def __init__(self, name, description, interface, frames, packets, runs,
+ inline_offset, drainer_offset, size, completion_counter=None,
+ fresh_per_run=False):
+ self.name = name
+ self.description = description
+ self.interface = interface
+ self.frame_size = size
+ self.frames = frames # pool of unique raw frames
+ self.packets_inline = packets.get("inline", 0)
+ self.packets_drainer = packets.get("drainer", 0)
+ self.runs = runs
+ self.inline_offset = inline_offset
+ self.drainer_offset = drainer_offset
+ self.completion_counter = completion_counter
+ # When true, each run uses a dedicated fresh slice of frames
+ # (required where processing is not idempotent, e.g. announces).
+ # When false, runs reuse the same pool with the duplicate filter
+ # reset in between.
+ self.fresh_per_run = fresh_per_run
+
+ def can_drainer(self):
+ return self.packets_drainer > 0 and self.completion_counter is not None
+
+ def n_for_mode(self, mode):
+ if mode == "inline": return self.packets_inline
+ if mode == "drainer": return self.packets_drainer
+
+
+class CountingDestination(RNS.Destination):
+ """A destination that counts successful deliveries, for use as a
+ completion counter in drainer mode."""
+
+ def __init__(self, *args, **kwargs):
+ super().__init__(*args, **kwargs)
+ self.delivered = 0
+
+ def receive(self, packet):
+ result = super().receive(packet)
+ if result:
+ self.delivered += 1
+ return result
+
+
+def _register_link(link):
+ with RNS.Transport.active_links_lock:
+ RNS.Transport.active_links.append(link)
+ if hasattr(RNS.Transport, "active_links_map"):
+ RNS.Transport.active_links_map[link.link_id] = link
+
+
+def _make_minimal_link(instance, interface):
+ """Constructs a minimal, active Link object suitable for benchmark delivery."""
+ import importlib
+ _link_mod = importlib.import_module("RNS.Link")
+
+ link = RNS.Link.__new__(RNS.Link)
+ link.mode = RNS.Link.MODE_DEFAULT
+ link.rtt = 0.1
+ link.mtu = RNS.Reticulum.MTU
+ link.establishment_cost = 0
+ link.establishment_rate = None
+ link.expected_rate = None
+ link.callbacks = _link_mod.LinkCallbacks()
+ link.resource_strategy = RNS.Link.ACCEPT_NONE
+ link.last_resource_window = None
+ link.last_resource_eifr = None
+ link.outgoing_resources = []
+ link.incoming_resources = []
+ link.pending_requests = []
+ link.last_inbound = 0
+ link.last_outbound = 0
+ link.last_keepalive = 0
+ link.last_proof = 0
+ link.last_data = 0
+ link.tx = 0
+ link.rx = 0
+ link.txbytes = 0
+ link.rxbytes = 0
+ link.rssi = None
+ link.snr = None
+ link.q = None
+ link.traffic_timeout_factor = RNS.Link.TRAFFIC_TIMEOUT_FACTOR
+ link.keepalive_timeout_factor = RNS.Link.KEEPALIVE_TIMEOUT_FACTOR
+ link.keepalive = RNS.Link.KEEPALIVE
+ link.stale_time = RNS.Link.STALE_TIME
+ link.watchdog_lock = False
+ link.status = RNS.Link.ACTIVE
+ link.activated_at = time.time()
+ link.type = RNS.Destination.LINK
+ link.owner = instance
+ link.initiator = False
+ link.expected_hops = 1
+ link.rebalanced = None
+ link.attached_interface = interface
+ link._channel = None
+ link._Link__remote_identity = None
+ link._Link__track_phy_stats = False
+ link.derived_key = os.urandom(64)
+ link.token = RNS.Cryptography.Token(link.derived_key)
+ link.link_id = os.urandom(RNS.Reticulum.TRUNCATED_HASHLENGTH//8)
+ link.hash = link.link_id
+ link.hexhash = link.link_id.hex()
+ from RNS.Cryptography.Proxies import Ed25519PrivateKeyProxy
+ link.sig_prv = Ed25519PrivateKeyProxy.from_private_bytes(link.derived_key[:32])
+
+ class _DestinationShim:
+ pass
+
+ link.destination = _DestinationShim()
+ link.destination.type = RNS.Destination.LINK
+ link.destination.proof_strategy = RNS.Destination.PROVE_NONE
+ _dest_mod = importlib.import_module("RNS.Destination")
+ link.destination.callbacks = _dest_mod.Callbacks()
+ link.destination.status = RNS.Link.ACTIVE
+ link.destination.last_outbound = 0
+ link.destination.tx = 0
+ link.destination.txbytes = 0
+ link.destination.attached_interface = interface
+ link.destination.mtu = RNS.Reticulum.MTU
+ link.destination.rssi = None
+ link.destination.snr = None
+ link.destination.q = None
+ link.destination.rtt = RNS.Link.TRAFFIC_TIMEOUT_MIN_MS/1000
+ link.destination.traffic_timeout_factor = 1.0
+
+ return link
+
+
+def _transit_single_frame(interface, dst_h, payload):
+ """One HEADER_2|TRANSPORT|SINGLE|DATA frame addressed to this instance."""
+ flags = (RNS.Packet.HEADER_2 << 6) | (RNS.Transport.TRANSPORT << 4) \
+ | (RNS.Destination.SINGLE << 2) | RNS.Packet.DATA
+ return (struct.pack("!B", flags) + struct.pack("!B", 1)
+ + RNS.Transport.identity.hash + dst_h
+ + bytes([RNS.Packet.NONE]) + payload)
+
+
+def _transit_link_frame(interface, link_id, payload):
+ """One HEADER_2|TRANSPORT|LINK|DATA frame addressed to this instance."""
+ flags = (RNS.Packet.HEADER_2 << 6) | (RNS.Transport.TRANSPORT << 4) \
+ | (RNS.Destination.LINK << 2) | RNS.Packet.DATA
+ return (struct.pack("!B", flags) + struct.pack("!B", 1)
+ + RNS.Transport.identity.hash + link_id
+ + bytes([RNS.Packet.NONE]) + payload)
+
+
+def _terminus_link_frame(link, payload):
+ """One HEADER_1|BROADCAST|LINK|DATA frame for a local active link,
+ encrypted with the link token."""
+ flags = (RNS.Packet.HEADER_1 << 6) | (RNS.Transport.BROADCAST << 4) \
+ | (RNS.Destination.LINK << 2) | RNS.Packet.DATA
+ ciphertext = link.encrypt(payload)
+ return (struct.pack("!B", flags) + struct.pack("!B", 0)
+ + link.link_id + bytes([RNS.Packet.NONE]) + ciphertext)
+
+
+def _payload(i, size, marker):
+ """Deterministic, unique payload of `size` bytes."""
+ return i.to_bytes(8, "big") + bytes([marker]) * (size - 8)
+
+
+def build_scenarios(instance, interface, interface_b, requested=None):
+ """Builds all benchmark scenarios and returns a dict name -> Scenario."""
+ scenarios = {}
+ runs = BENCHMARK_CONFIG["runs"]
+
+ def _slices(packets, fresh=False):
+ inline_n = packets.get("inline", 0)
+ drainer_n = packets.get("drainer", 0)
+ if fresh:
+ # Fresh frames per run for both modes
+ total = (inline_n + drainer_n) * runs
+ return total, inline_n, drainer_n, 0, inline_n * runs
+ else:
+ # One re-usable pool of the largest mode's frame count
+ total = max(inline_n, drainer_n)
+ return total, inline_n, drainer_n, 0, 0
+
+ # --- SINGLE transit relay, several payload sizes -----------------------
+ sizes = [("135", 135), ("475", 475), ("1024", 1024), ("16384", 16384)]
+ for suffix, size in sizes:
+ name = f"transit_single_{suffix}"
+ total, inline_n, drainer_n, off_i, off_d = _slices(DEFAULT_PACKETS[name])
+ dst_h = RNS.Cryptography.hkdf(length=16,
+ derive_from=f"T{suffix}dst".encode(),
+ salt=b"bench", context=None)
+ next_hop = RNS.Cryptography.hkdf(length=16,
+ derive_from=f"T{suffix}nh".encode(),
+ salt=b"bench", context=None)
+ now = time.time()
+ with RNS.Transport.path_table_lock:
+ RNS.Transport.path_table[dst_h] = [
+ now, next_hop, 3, now + 3600, [], interface, bytes(32)
+ ]
+ frames = [_transit_single_frame(interface, dst_h,
+ _payload(i, size, 0x57))
+ for i in range(total)]
+ scenarios[name] = Scenario(
+ name,
+ f"SINGLE transit relay, {size} B payload, 3-hop path",
+ interface, frames, DEFAULT_PACKETS[name], runs,
+ off_i, off_d, size,
+ completion_counter=lambda: RNS.Transport.tx_packets,
+ )
+
+ # --- SINGLE transit relay, final hop (header strip) --------------------
+ name = "transit_single_final"
+ total, inline_n, drainer_n, off_i, off_d = _slices(DEFAULT_PACKETS[name])
+ dst_h = RNS.Cryptography.hkdf(length=16, derive_from=b"TFdst",
+ salt=b"bench", context=None)
+ next_hop = RNS.Cryptography.hkdf(length=16, derive_from=b"TFnh",
+ salt=b"bench", context=None)
+ now = time.time()
+ with RNS.Transport.path_table_lock:
+ RNS.Transport.path_table[dst_h] = [
+ now, next_hop, 1, now + 3600, [], interface, bytes(32)
+ ]
+ frames = [_transit_single_frame(interface, dst_h,
+ _payload(i, 135, 0x46))
+ for i in range(total)]
+ scenarios[name] = Scenario(
+ name,
+ "SINGLE transit relay, final hop, transport header strip, 135 B",
+ interface, frames, DEFAULT_PACKETS[name], runs,
+ off_i, off_d, 135,
+ completion_counter=lambda: RNS.Transport.tx_packets,
+ )
+
+ # --- LINK transit relay (cross-interface) ------------------------------
+ sizes = [("135", 135), ("475", 475), ("1024", 1024), ("16384", 16384)]
+ for suffix, size in sizes:
+ name = f"transit_link_{suffix}"
+
+ total, inline_n, drainer_n, off_i, off_d = _slices(DEFAULT_PACKETS[name])
+ link_id = RNS.Cryptography.hkdf(length=16, derive_from=b"Llink",
+ salt=b"bench", context=None)
+ now = time.time()
+ with RNS.Transport.link_table_lock:
+ # timestamp, next-hop transport id, outbound iface, remaining hops,
+ # received-on iface, taken hops, original destination hash,
+ # validated, proof timeout
+ RNS.Transport.link_table[link_id] = [
+ now, RNS.Transport.identity.hash, interface_b, 2,
+ interface, 2, os.urandom(16), True, now + 60
+ ]
+ frames = [_transit_link_frame(interface, link_id,
+ _payload(i, size, 0x4C))
+ for i in range(total)]
+ scenarios[name] = Scenario(
+ name,
+ f"LINK transit relay via link table, cross-interface, {size} B",
+ interface, frames, DEFAULT_PACKETS[name], runs,
+ off_i, off_d, size,
+ completion_counter=lambda: RNS.Transport.tx_packets,
+ )
+
+ # --- LINK terminus delivery --------------------------------------------
+ name = "terminus_link"
+ total, inline_n, drainer_n, off_i, off_d = _slices(DEFAULT_PACKETS[name])
+ link = _make_minimal_link(instance, interface)
+ _register_link(link)
+ frames = [_terminus_link_frame(link, _payload(i, 135, 0x6C))
+ for i in range(total)]
+ scenarios[name] = Scenario(
+ name,
+ "LINK terminus delivery, token decrypt, no app callback, 135 B",
+ interface, frames, DEFAULT_PACKETS[name], runs,
+ off_i, off_d, 135,
+ completion_counter=lambda: link.rx,
+ )
+
+ # --- SINGLE local delivery ---------------------------------------------
+ name = "terminus_single"
+ total, inline_n, drainer_n, off_i, off_d = _slices(DEFAULT_PACKETS[name])
+ identity = RNS.Identity()
+ # The Destination initialiser automatically registers IN destinations
+ # with the transport core.
+ destination = CountingDestination(identity, RNS.Destination.IN,
+ RNS.Destination.SINGLE,
+ "bench", "terminus")
+ frames = []
+ for i in range(total):
+ packet = RNS.Packet(destination, _payload(i, 135, 0x53),
+ RNS.Packet.DATA, create_receipt=False)
+ packet.pack()
+ frames.append(packet.raw)
+ scenarios[name] = Scenario(
+ name,
+ "SINGLE local delivery, ephemeral-key decrypt, 135 B",
+ interface, frames, DEFAULT_PACKETS[name], runs,
+ off_i, off_d, 135,
+ completion_counter=lambda: destination.delivered,
+ )
+
+ # --- Announce ingress ----------------------------------------------------
+ name = "announce_ingress"
+ total, inline_n, drainer_n, off_i, off_d = _slices(DEFAULT_PACKETS[name],
+ fresh=True)
+ name_hash = RNS.Identity.full_hash(b"rns.throughput.bench")[
+ :RNS.Identity.NAME_HASH_LENGTH//8]
+ frames = []; tsize = 0
+ for i in range(total):
+ ann_identity = RNS.Identity()
+ dst_h = RNS.Identity.full_hash(name_hash + ann_identity.hash)[
+ :RNS.Reticulum.TRUNCATED_HASHLENGTH//8]
+ random_hash = os.urandom(5) + int(time.time() + i).to_bytes(5, "big")
+ signed_data = dst_h + ann_identity.get_public_key() + name_hash \
+ + random_hash + b""
+ signature = ann_identity.sign(signed_data)
+ announce_data = ann_identity.get_public_key() + name_hash \
+ + random_hash + b"" + signature
+ flags = (RNS.Packet.HEADER_1 << 6) \
+ | (RNS.Destination.SINGLE << 2) | RNS.Packet.ANNOUNCE
+ fbs = struct.pack("!B", flags) + struct.pack("!B", 0) + dst_h + bytes([RNS.Packet.NONE]) + announce_data
+ frames.append(fbs)
+ tsize += len(fbs)
+
+ scenarios[name] = Scenario(
+ name,
+ "Announce ingress, fresh destinations, validation + path insert",
+ interface, frames, DEFAULT_PACKETS[name], runs,
+ off_i, off_d, int(tsize/total),
+ completion_counter=lambda: len(RNS.Transport.path_table),
+ fresh_per_run=True,
+ )
+
+ # --- Outbound insertion into transport ----------------------------------
+ name = "outbound_path"
+ packets = DEFAULT_PACKETS[name]
+ inline_n = packets.get("inline", 0)
+ total = inline_n # pool is re-used across runs
+ dst_h = RNS.Cryptography.hkdf(length=16, derive_from=b"Odst",
+ salt=b"bench", context=None)
+ next_hop = RNS.Cryptography.hkdf(length=16, derive_from=b"Onh",
+ salt=b"bench", context=None)
+ now = time.time()
+ with RNS.Transport.path_table_lock:
+ RNS.Transport.path_table[dst_h] = [
+ now, next_hop, 3, now + 3600, [], interface, bytes(32)
+ ]
+ remote_id = RNS.Identity(create_keys=False)
+ remote_id.load_public_key(os.urandom(RNS.Identity.KEYSIZE//8))
+ outbound_destination = RNS.Destination(remote_id, RNS.Destination.OUT,
+ RNS.Destination.SINGLE,
+ "bench", "outbound")
+ outbound_destination.hash = dst_h
+ outbound_destination.hexhash = dst_h.hex()
+ frames = []
+ for i in range(total):
+ packet = RNS.Packet(outbound_destination, _payload(i, 135, 0x4F),
+ RNS.Packet.DATA, create_receipt=False)
+ packet.pack()
+ frames.append(packet)
+ outbound = Scenario(
+ name,
+ "Outbound insertion into transport, known 3-hop path, 135 B",
+ interface, frames, packets, runs,
+ 0, 0, 135,
+ completion_counter=None,
+ )
+ outbound.kind = "outbound"
+ scenarios[name] = outbound
+
+ if requested is not None:
+ missing = [s for s in requested if s not in scenarios]
+ if missing:
+ raise KeyError(f"Unknown scenario(s): {', '.join(missing)}. "
+ f"Available: {', '.join(sorted(scenarios))}")
+ scenarios = {name: scenarios[name] for name in requested}
+
+ return scenarios
+
+
+def _reset_transport_state():
+ RNS.Transport.packet_hashlist = set()
+ RNS.Transport.packet_hashlist_prev = set()
+ RNS.Transport.reverse_table = {}
+
+
+def _percentile(sorted_values, p):
+ if not sorted_values:
+ return 0.0
+ index = min(len(sorted_values) - 1, int(len(sorted_values) * p))
+ return sorted_values[index]
+
+
+def _feed(frame, interface):
+ RNS.Transport.preprocess_inbound(frame, interface)
+
+def _bench_inline(interface, frames, n, runs, offset, fresh_per_run=False):
+ """Synchronous benchmark: process each frame in the calling thread,
+ bypassing the inbound queue. Returns (run_means_us, samples_us)."""
+ run_means = []
+ all_samples = []
+
+ # Bypass the inbound queue so processing happens entirely in the
+ # calling thread; restore the previous setting afterwards.
+ previous_queue_state = RNS.Transport.USE_INBOUND_QUEUE
+ RNS.Transport.USE_INBOUND_QUEUE = False
+ try:
+ for r in range(runs):
+ _reset_transport_state()
+ gc.collect()
+ gc.disable()
+
+ base = offset + (r * n if fresh_per_run else 0)
+ t0 = time.perf_counter()
+ prev = None
+ samples = []
+ step = INLINE_SAMPLE_STEP
+ next_sample = step
+ for i in range(n):
+ _feed(frames[base + i], interface)
+ if i == next_sample:
+ now = time.perf_counter()
+ if prev is not None:
+ samples.append((now - prev) / step)
+ prev = now
+ next_sample += step
+ dt = time.perf_counter() - t0
+ gc.enable()
+
+ run_means.append(dt / n * 1e6)
+ all_samples.extend(samples)
+ finally:
+ RNS.Transport.USE_INBOUND_QUEUE = previous_queue_state
+
+ return run_means, [s * 1e6 for s in all_samples]
+
+
+def _bench_drainer(scenario, n, runs):
+ """Feed frames through the inbound queue from the calling thread
+ while the drainer thread processes them. Backpressure is applied
+ via the scenario completion counter, so no frames are dropped and
+ the pipeline stays saturated."""
+ run_means = []
+
+ # Ensure the inbound queue and drainer are used; restore afterwards.
+ previous_queue_state = RNS.Transport.USE_INBOUND_QUEUE
+ RNS.Transport.USE_INBOUND_QUEUE = True
+ try:
+ for r in range(runs):
+ _reset_transport_state()
+ gc.collect()
+ gc.disable()
+
+ base = scenario.drainer_offset \
+ + (r * n if scenario.fresh_per_run else 0)
+ baseline = scenario.completion_counter()
+ fed = 0
+ t0 = time.perf_counter()
+ chunk = DRAINER_CHUNK
+ while fed < n:
+ batch = min(chunk, n - fed)
+ for i in range(fed, fed + batch):
+ _feed(scenario.frames[base + i], scenario.interface)
+ fed += batch
+ target = baseline + fed
+ while scenario.completion_counter() < target:
+ time.sleep(0.0002)
+ dt = time.perf_counter() - t0
+ gc.enable()
+
+ run_means.append(dt / n * 1e6)
+ finally:
+ RNS.Transport.USE_INBOUND_QUEUE = previous_queue_state
+
+ return run_means
+
+
+def _bench_outbound(packets, n, runs, offset):
+ """Synchronous benchmark of the outbound path: insert each packet
+ into transport for a known path via Transport.outbound(). The packet
+ pool is re-used across runs."""
+ run_means = []
+ for r in range(runs):
+ gc.collect()
+ gc.disable()
+ t0 = time.perf_counter()
+ for i in range(n):
+ RNS.Transport.outbound(packets[offset + i])
+ dt = time.perf_counter() - t0
+ gc.enable()
+ run_means.append(dt / n * 1e6)
+ return run_means
+
+
+# ---------------------------------------------------------------------------
+# Output helpers
+# ---------------------------------------------------------------------------
+
+def _fmt_us(value):
+ return f"{value:>10.2f}"
+
+
+def _fmt_pps(value):
+ return f"{value:>12,.0f}"
+
+
+def print_environment_header():
+ tree_fastpath = "present" if hasattr(RNS.Transport, "_fastpath") else "absent"
+ print("=" * 40)
+ print("Reticulum Transport Throughput Benchmark")
+ print("=" * 40)
+ print(f" RNS version : {RNS.__version__}")
+ print(f" Mode : {'compiled' if RNS.compiled else 'interpreted'}")
+ print(f" crypto : {RNS.Cryptography.backend()}")
+ print(f" transport : enabled")
+ print(f" fast path : {tree_fastpath} (informational)")
+ print(f" python : {platform.python_version()}")
+ print(f" platform : {platform.platform()}")
+ print(f" machine : {platform.machine()}, cpus: {os.cpu_count()}")
+ print()
+
+
+def print_scenario_table(scenario, results):
+ """results: dict mode -> dict(runs, mean_us, pps, stdev, p50, p95)."""
+ print(f"Scenario: {scenario.name} - {scenario.description}")
+ print(f" {'mode':<9}{'n':>8}{'runs':>6}{'mean ยตs':>13}{'pps':>16}"
+ f"{'p50':>11}{'p95':>11}")
+ print(" " + "-" * 75)
+ for mode in ("inline", "drainer"):
+ if mode not in results:
+ continue
+ r = results[mode]
+ n = scenario.n_for_mode(mode)
+ p50 = f"{r['p50']:>9.2f} ยตs" if r["p50"] is not None else "-"
+ p95 = f"{r['p95']:>9.2f} ยตs" if r["p95"] is not None else "-"
+ spread = ""
+ if r.get("stdev") is not None:
+ spread = f" ยฑ{r['stdev']:.2f}"
+ if mode == "inline": mode = "direct"
+ print(f" {mode:<9}{n:>8}{r['runs']:>6}{_fmt_us(r['mean_us']):>13}"
+ f"{_fmt_pps(r['pps']):>16}{p50:>11}{p95:>11} "
+ f"({r['runs']} runs, median{spread})")
+
+ print()
+ for mode in ("inline", "drainer"):
+ if mode not in results:
+ continue
+
+ r = results[mode]
+ tp = r['pps']*scenario.frame_size*8
+ print(f"{mode:<8} : {RNS.prettyspeed(tp)}")
+
+ print()
+
+
+def print_pps_matrix(rows):
+ """rows: list of (scenario_name, inline_pps, drainer_pps, fsize)"""
+ if not rows:
+ return
+ print("-" * 72)
+ print("Transport Core Throughput - PPS matrix (median of runs)")
+ print(f" {'scenario':<22}{'direct':>14}{'drainer':>14}")
+ print(" " + "-" * 70)
+ for name, inline_pps, drainer_pps, fsize in rows:
+ i = f"{inline_pps:>12,.0f}" if inline_pps else "-"
+ d = f"{drainer_pps:>12,.0f}" if drainer_pps else "-"
+ itp = RNS.prettyspeed(inline_pps*fsize*8) if inline_pps else "-"
+ dtp = RNS.prettyspeed(drainer_pps*fsize*8) if drainer_pps else "-"
+ print(f" {name:<22}{i:>14}{d:>14}{itp:>16} / {dtp}")
+ print()
+
+
+# ---------------------------------------------------------------------------
+# Test suite
+# ---------------------------------------------------------------------------
+
+_instance = None
+_interfaces = None
+_scenarios = None
+_config_dir = None
+
+
+def _start_transport():
+ global _instance, _interfaces, _scenarios, _config_dir
+
+ _config_dir = tempfile.mkdtemp(prefix="rns-transport-bench-")
+ config_dir = _config_dir
+ with open(os.path.join(config_dir, "config"), "w") as fh:
+ fh.write(
+ "[reticulum]\n"
+ " enable_transport = yes\n"
+ " share_instance = No\n"
+ " panic_on_interface_error = No\n"
+ "\n"
+ "[logging]\n"
+ " loglevel = 0\n"
+ "\n"
+ "[interfaces]\n"
+ )
+
+ _instance = RNS.Reticulum(configdir=config_dir,
+ loglevel=RNS.LOG_CRITICAL,
+ logdest=lambda *a, **k: None)
+
+ while not RNS.Transport.ready:
+ time.sleep(0.05)
+
+ interface = BenchmarkInterface("line-a")
+ interface_b = BenchmarkInterface("line-b")
+ RNS.Transport.add_interface(interface)
+ RNS.Transport.add_interface(interface_b)
+ RNS.Transport.prioritize_interfaces()
+ _interfaces = (interface, interface_b)
+
+ requested = BENCHMARK_CONFIG["scenarios"]
+ _scenarios = build_scenarios(_instance, interface, interface_b,
+ requested=requested)
+
+
+class TestTransportThroughput(unittest.TestCase):
+
+ @classmethod
+ def setUpClass(cls):
+ if _scenarios is None:
+ _start_transport()
+
+ @classmethod
+ def tearDownClass(cls):
+ if RNS.Transport._should_run:
+ RNS.Transport.exit_handler()
+ if _config_dir is not None:
+ import shutil
+ shutil.rmtree(_config_dir, ignore_errors=True)
+
+ def _run_scenario_matrix(self, scenario_names):
+ rows = []
+ # Only include scenarios selected via the command line
+ scenario_names = [name for name in scenario_names
+ if name in _scenarios]
+ for name in scenario_names:
+ scenario = _scenarios[name]
+ results = {}
+
+ mode_filter = BENCHMARK_CONFIG["mode"]
+ runs = BENCHMARK_CONFIG["runs"]
+
+ if scenario.kind == "outbound":
+ # The outbound path is synchronous; it only has an
+ # inline-equivalent measurement, and follows the mode
+ # filter's "inline" selector.
+ if mode_filter in ("inline", "both"):
+ run_means = _bench_outbound(scenario.frames,
+ scenario.packets_inline,
+ runs, 0)
+ mean_us = statistics.median(run_means)
+ results["inline"] = {
+ "runs": runs,
+ "mean_us": mean_us,
+ "pps": 1e6 / mean_us,
+ "stdev": statistics.stdev(run_means) if len(run_means) > 2 else None,
+ "p50": None,
+ "p95": None,
+ }
+ else:
+ if mode_filter in ("inline", "both") and scenario.packets_inline:
+ n = scenario.packets_inline
+ run_means, samples = _bench_inline(scenario.interface,
+ scenario.frames, n, runs,
+ scenario.inline_offset,
+ scenario.fresh_per_run)
+ mean_us = statistics.median(run_means)
+ stdev = statistics.stdev(run_means) if len(run_means) > 2 else None
+ sorted_samples = sorted(samples)
+ results["inline"] = {
+ "runs": runs,
+ "mean_us": mean_us,
+ "pps": 1e6 / mean_us,
+ "stdev": stdev,
+ "p50": _percentile(sorted_samples, 0.50),
+ "p95": _percentile(sorted_samples, 0.95),
+ }
+
+ if mode_filter in ("drainer", "both") and scenario.can_drainer():
+ n = scenario.packets_drainer
+ run_means = _bench_drainer(scenario, n, runs)
+ mean_us = statistics.median(run_means)
+ stdev = statistics.stdev(run_means) if len(run_means) > 2 else None
+ results["drainer"] = {
+ "runs": runs,
+ "mean_us": mean_us,
+ "pps": 1e6 / mean_us,
+ "stdev": stdev,
+ "p50": None,
+ "p95": None,
+ }
+
+ if results:
+ print_scenario_table(scenario, results)
+ rows.append((name,
+ results.get("inline", {}).get("pps"),
+ results.get("drainer", {}).get("pps"),
+ scenario.frame_size))
+
+ print_pps_matrix(rows)
+
+ def test_01_transit_throughput(self):
+ print("")
+ print_environment_header()
+ self._run_scenario_matrix([
+ "transit_single_135",
+ "transit_single_475",
+ "transit_single_1024",
+ "transit_single_16384",
+ "transit_single_final",
+ "transit_link_135",
+ "transit_link_475",
+ "transit_link_1024",
+ "transit_link_16384",
+ ])
+
+ def test_02_delivery_throughput(self):
+ print("")
+ self._run_scenario_matrix([
+ "terminus_link_135",
+ "terminus_link_475",
+ "terminus_link_1024",
+ "terminus_link_16384",
+ "terminus_single",
+ "announce_ingress",
+ "outbound_path",
+ ])
+
+
+def _usage():
+ return (
+ "\nUsage: python3 tests/transport_throughput.py [options]\n"
+ "\n"
+ "Options:\n"
+ " -s, --scenario NAME Run only the named scenario (repeatable)\n"
+ " --mode MODE Measurement mode: inline, drainer or both\n"
+ " (default: both)\n"
+ " --runs N Measurement runs per scenario/mode, median\n"
+ " is reported (default: 3)\n"
+ " --list-scenarios List available scenarios and exit\n"
+ " -h, --help Show this help and exit\n"
+ )
+
+
+if __name__ == "__main__":
+ argv = sys.argv[1:]
+ rest = []
+ i = 0
+ while i < len(argv):
+ arg = argv[i]
+ if arg in ("-s", "--scenario"):
+ BENCHMARK_CONFIG["scenarios"] = BENCHMARK_CONFIG["scenarios"] or []
+ BENCHMARK_CONFIG["scenarios"].append(argv[i + 1])
+ i += 2
+ elif arg == "--mode":
+ mode = argv[i + 1]
+ if mode not in ("inline", "drainer", "both"):
+ raise SystemExit(f"Invalid mode '{mode}'" + _usage())
+ BENCHMARK_CONFIG["mode"] = mode
+ i += 2
+ elif arg == "--runs":
+ BENCHMARK_CONFIG["runs"] = max(1, int(argv[i + 1]))
+ i += 2
+ elif arg == "--list-scenarios":
+ print("\nAvailable scenarios:")
+ for name in sorted(SCENARIO_DESCRIPTIONS):
+ print(f" {name:<22} {SCENARIO_DESCRIPTIONS[name]}")
+ raise SystemExit(0)
+ elif arg in ("-h", "--help"):
+ raise SystemExit(_usage())
+ else:
+ rest.append(arg)
+ i += 1
+
+ unittest.main(argv=[sys.argv[0]] + rest, verbosity=2)
+
Served by rngit 1.5.2 - Generated in 0.05s